package com.atguigu.edu.realtime.dwd.db.app;


import com.atguigu.edu.realtime.common.base.BaseSqlApp;
import com.atguigu.edu.realtime.common.constant.Constant;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import org.apache.flink.configuration.Configuration;

public class DwdTradeOrderDetail extends BaseSqlApp {
    public static void main(String[] args) {
        new DwdTradeOrderDetail().start(
                10016, 4, Constant.TOPIC_DWD_TRADE_ORDER_DETAIL, Constant.TOPIC_DWD_TRADE_ORDER_DETAIL
        );
    }

    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tableEnv) {
        getTopicDbTable(tableEnv,"DwdTradeOrderDetail");
        //Table table1 = tableEnv.sqlQuery("select * from topic_db");
        //table1.execute().print();
        //过滤order_detail——insert
        Table orderDetail = tableEnv.sqlQuery("select \n" +
                "`table`,\n" +
                "data['id'] id,\n" +
                "data['user_id'] user_id,\n" +
                "data['order_id'] order_id,\n" +
                "data['course_id'] course_id,\n" +
                "data['course_name'] course_name,\n" +
                "data['create_time'] create_time,\n" +
                "data['original_amount'] origin_amount,\n" +
                "data['coupon_amount'] coupon_amount,\n" +
                "data['final_amount'] final_amount,\n" +
                "data['update_time'] update_time,\n" +
                "ts \n" +
                "from topic_db \n" +
                "where  `table`='order_detail' \n" +
                "and `type`='insert' ");
        tableEnv.createTemporaryView("order_detail", orderDetail);
        Table orderInfo = tableEnv.sqlQuery("select \n" +
                "`table`,\n" +
                "data['id'] id,\n" +
                "data['user_id'] user_id,\n" +
                "data['create_time'] create_time,\n" +
                "data['origin_amount'] original_amount,\n" +
                "data['coupon_reduce'] coupon_amount,\n" +
                "data['final_amount'] final_amount,\n" +
                "data['order_status'] order_status,\n" +
                "data['update_time'] update_time,\n" +
                "data['out_trade_no'] out_trade_no,\n" +
                "data['trade_body'] trade_body,\n" +
                "data['session_id'] session_id,\n" +
                "data['province_id'] province_id,\n" +
                "data['expire_time'] expire_time,\n" +
                "ts \n" +
                "from topic_db \n" +
                "where  `table`='order_info' \n" +
                "and `type`='insert' ");
        tableEnv.createTemporaryView("order_info", orderInfo);
        Table table = tableEnv.sqlQuery("select \n" +
                "od.id              id,\n" +
                "od.order_id        order_id,\n" +
                "oi.user_id         user_id,\n" +
                "oi.order_status    order_status,\n" +
                "oi.out_trade_no    out_trade_no,\n" +
                "oi.trade_body      trade_body,\n" +
                "oi.session_id      session_id,\n" +
                "oi.province_id     province_id,\n" +
                "od.course_id       course_id,\n" +
                "od.course_name     course_name,\n" +
                "od.origin_amount   origin_amount,\n" +
                "od.coupon_amount   coupon_amount,\n" +
                "od.final_amount    final_amount,\n" +
                "od.create_time     create_time,\n" +
                "od.update_time     update_time,\n" +
                "od.ts ts\n" +
                "from \n" +
                "order_detail od\n" +
                "join order_info oi\n" +
                "on od.order_id = oi.id");
       // table.execute().print();
        tableEnv.executeSql(
                "create table dwd_trade_order_detail(\n" +
                        "id string,\n" +
                        "order_id string,\n" +
                        "user_id string,\n" +
                        "order_status string,\n" +
                        "out_trade_no string,\n" +
                        "trade_body string,\n" +
                        "session_id string,\n" +
                        "province_id string,\n" +
                        "course_id string,\n" +
                        "course_name string,\n" +
                        "origin_amount string,\n" +
                        "coupon_amount string,\n" +
                        "final_amount string,\n" +
                        "create_time string,\n" +
                        "update_time string ,\n" +
                        "ts bigint,\n" +
                        "primary key(id) not enforced " +
                        ")" + "WITH (\n" +
                        "  'connector' = 'upsert-kafka',\n" +
                        "  'topic' = '" + Constant.TOPIC_DWD_TRADE_ORDER_DETAIL + "',\n" +
                        "  'properties.bootstrap.servers' = '" + Constant.KAFKA_BROKERS + "',\n" +
                        "  'key.format' = 'json',\n" +
                        "  'value.format' = 'json'\n" +
                        ")"
        );
        table.executeInsert(Constant.TOPIC_DWD_TRADE_ORDER_DETAIL);
    }
}
